<?php

ini_set('display_errors', 1);
error_reporting(-1);
require_once '/data/wwwroot/g_vendor/autoload.php';

use Biz\service\Event;
use Doraemon\tools\Log;
use Doraemon\amqp\Queue as aQueue;
use Biz\model\eventlog\General as mGeneral;

$queueNames = [];
$forceConnect = false;
$params = $argv;

$queueNames = array_map(function ($v) {
    return [
        'exchange' => $v::EVENT_PUSH_EXCHANGE,
        'default_queue' => $v::EVENT_PUSH_DEFAULT_QUEUE
    ];
}, Event::EVENT_SUB);

if (empty($argv[1])) {
    echo "请传入要同步的事件参数,一次只能填写一个\n";
    echo "参数列表如下\n";
    foreach ($queueNames as $key => $value) {
        echo "\t" . $key . "\n";
        echo "\t\t 示例:php " . $params[0] . " " . $key . "\n";
    }
    exit;
} else {

    if (empty($queueNames[$params[1]])) {
        exit("参数传递错误");
    }

    list($exchange, $queueName) = array_values($queueNames[$params[1]]);

}



$callback = function ($message){
    $data = json_decode($message->body,true);

    if(!isset($data['body'])){
        Event::ack($message);
        return false;
    }

    $eventTag = $data['header']['event_tag'];
    $eventId = $data['header']['event_id'];
    $msgId = $data['header']['msg_id'];
    $msgAddtime = $data['header']['addtime'];

    $insertData = $data['body'];

    try{
        $id = mGeneral::createLog($eventTag,$eventId,$msgId,$insertData,$msgAddtime);

        if($id > 0){
            Event::ack($message);

            Log::event_syncmqconsume_info('同步 MQ 事件到 mysql 的任务 正常', [
                'data'=>$data,
            ]);
        }else{
            throw new Exception('消息落地出错');
        }
    }catch (\Exception $e){
        Log::event_syncmqconsume_error('同步 MQ 事件到 mysql 的任务 出现异常', [
            'data'=>$data,
            'e_msg' => $e->getMessage(),
            'e_file' => $e->getFile(),
            'e_line' => $e->getLine()
        ]);

        //后台任务,处理取消订单成功的事件失败
        aQueue::insertQueue("queue/Wechatnotice/toTag", [
            "totag"=>\Biz\pockets\constInterface\BizWechatTagConfig::TAG_BIZ_RD_SYSTEM,
            "content" => vsprintf('同步 MQ 事件到 mysql 的任务 出现异常,info=%s', [
                json_encode([
                    'data'=>$data,
                    'e_msg' => $e->getMessage(),
                    'e_file' => $e->getFile(),
                    'e_line' => $e->getLine()
                ], JSON_UNESCAPED_UNICODE),
            ])
        ], aQueue::QUEUE_BIZ);

        Event::ack($message);

    }


};

while (true) {
    try {

        $channel = Event::getChannel($forceConnect, 'syncmq');

        $channel->queue_declare($queueName, false, true, false, false);

        $channel->exchange_declare($exchange,'topic', false, true, false);

        $channel->queue_bind($queueName,$exchange);


        $channel->basic_qos(null, 1, null);

        $channel->basic_consume($queueName, '', false, false, false, false,$callback);

        $forceConnect = false;

        while (count($channel->callbacks)) {
            $channel->wait();
        }

    } catch (\Exception $e) {

        Log::event_syncmqconsume_error('同步 MQ 事件到 mysql 的任务 出现异常', [
            'e_msg' => $e->getMessage(),
            'e_file' => $e->getFile(),
            'e_line' => $e->getLine()
        ]);

        //后台任务,处理取消订单成功的事件失败
        aQueue::insertQueue("queue/Wechatnotice/toTag", [
            "totag"=>\Biz\pockets\constInterface\BizWechatTagConfig::TAG_BIZ_RD_SYSTEM,
            "content" => vsprintf('同步 MQ 事件到 mysql 的任务 出现异常,info=%s', [
                json_encode([
                    'e_msg' => $e->getMessage(),
                    'e_file' => $e->getFile(),
                    'e_line' => $e->getLine()
                ], JSON_UNESCAPED_UNICODE),
            ])
        ], aQueue::QUEUE_BIZ);

        $forceConnect = true;

        sleep(3);

        continue;
    }

}

